/*
 * syswaitqueue.hpp
 *
 *  Created on: 2016年11月11日
 *      Author: Dylan.Gao
 */

#ifndef _DM_OS_SYSWAITQUEUE_HPP_
#define _DM_OS_SYSWAITQUEUE_HPP_

#include <dm/types.hpp>
#include <dm/ringpos_v.hpp>
#include <dm/ringpos.hpp>
#include <dm/os/mustsharablelock.hpp>
#include <dm/os/mustlock.hpp>
#include <boost/interprocess/sync/sharable_lock.hpp>
#include <boost/interprocess/sync/interprocess_sharable_mutex.hpp>
#include <boost/interprocess/sync/interprocess_condition_any.hpp>
#include <boost/date_time/posix_time/ptime.hpp>

namespace dm{
namespace os{

template< typename T,dm::uint Size>
class CSysWaitQueue{
	CSysWaitQueue( const CSysWaitQueue& );
	CSysWaitQueue& operator=( const CSysWaitQueue& );
public:
	typedef dm::uint16 pos_t;
	typedef dm::uint16 cycle_t;
	typedef dm::uint32 size_t;

	typedef boost::interprocess::interprocess_sharable_mutex lockshared_t;
	typedef boost::interprocess::interprocess_condition_any cond_t;
	typedef boost::interprocess::sharable_lock<lockshared_t> sslock_t;
	typedef boost::posix_time::ptime time_t;

	typedef dm::os::CMustSharableLock<lockshared_t> msharelock_shared_t;
	typedef dm::os::CMustLock<lockshared_t> msharelock_exclusive_t;

	typedef dm::CRingPosV<pos_t,pos_t,Size,cycle_t> ringpos_v_t;
	typedef dm::CRingPos<pos_t,pos_t,Size,cycle_t> ringpos_t;

public:
	CSysWaitQueue();
	~CSysWaitQueue(){}

	void push( const T& e );
	void push( const T& e,ringpos_t& p );

	bool pop( ringpos_t& p,T& e );

	bool wait( const time_t& t );
	void wait( ringpos_t& p,T& e );

	inline bool popAndCheckOver( ringpos_t& p,bool& over,T& e ){
		over = (!getCurPos().isNoMoreThanACycle(p));
		return pop(p,e);
	}

	/**
	* 等待，并检测是否溢出
	* 返回值：是否超时
	* - true 超时
	* - false 未超时，获取到记录
	*/
	inline bool waitAndCheckOver(ringpos_t& p, bool& over, T& e, int secs=-1);

	ringpos_t getCurPos(){
		msharelock_shared_t ml(lock);
		ringpos_t rt;
		rt = pos;
		return rt;
	}

	inline pos_t getCurSize()const{
		return pos.getSize();
	}

private:
	inline void movePosAndNotify(){
		++pos;
		cond.notify_all();
	}

private:
	lockshared_t lock;
	cond_t cond;
	ringpos_v_t pos;

	T elems[Size];
};

template< typename T,dm::uint Size>
CSysWaitQueue<T,Size>::CSysWaitQueue():lock(),cond(),pos(){
}

template< typename T,dm::uint Size>
void CSysWaitQueue<T,Size>::push( const T& e ){
	msharelock_exclusive_t ml(lock);
	elems[pos.getPos()] = e;
	movePosAndNotify();
}

template< typename T,dm::uint Size>
void CSysWaitQueue<T,Size>::push( const T& e,CSysWaitQueue<T,Size>::ringpos_t& p ){
	msharelock_exclusive_t ml(lock);
	pos.getRingPos(p);
	elems[pos.getPos()] = e;

	movePosAndNotify();
}

template< typename T,dm::uint Size>
bool CSysWaitQueue<T,Size>::pop( CSysWaitQueue<T,Size>::ringpos_t& p,T& e ){
	if( pos==p )
		return false;

	e = elems[p.getPos()];
	++p;

	return true;
}

template< typename T,dm::uint Size>
bool CSysWaitQueue<T,Size>::wait(const time_t& ptime ){
	sslock_t sl(lock);
	return !cond.timed_wait(sl,ptime);
}

template< typename T,dm::uint Size>
void CSysWaitQueue<T,Size>::wait( CSysWaitQueue<T,Size>::ringpos_t& p,T& e ){
	while( !pop(p,e) )
		wait(boost::posix_time::second_clock::universal_time()+boost::posix_time::seconds(1));
}

template< typename T, dm::uint Size>
bool CSysWaitQueue<T, Size>::waitAndCheckOver(CSysWaitQueue<T, Size>::ringpos_t& p, bool& over, T& e, int secs) {
	for (int i = 0; i != secs; ++i) {
		if (popAndCheckOver(p, over, e))
			return false;
		wait(boost::posix_time::second_clock::universal_time() + boost::posix_time::seconds(1));
	}
	return true;
}

}
}

#define DM_OS_TEMPLATE_INIT_SYSWAITQUEUE(PORT_TYPE,T,Size) \
template class PORT_TYPE boost::interprocess::sharable_lock< dm::os::CSysWaitQueue<T,Size>::lockshared_t> >; \
template class PORT_TYPE dm::os::CMustSharableLock< dm::os::CSysWaitQueue<T,Size>::lockshared_t>; \
template class PORT_TYPE dm::os::CSysWaitQueue<T,Size>;

#endif /* INCLUDE_DM_OS_SYSWAITQUEUE_HPP_ */
